package com.aispeech.asr.apis.ssc.service;

import static org.springframework.data.mongodb.core.query.Criteria.where;
import static org.springframework.data.mongodb.core.query.Query.query;

import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import org.springframework.context.ApplicationContext;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.stereotype.Component;

import com.aispeech.asr.apis.ssc.config.AppConfig;
import com.aispeech.asr.apis.ssc.constants.Progress;
import com.aispeech.asr.apis.ssc.data.domain.TaskDoc;
import com.aispeech.asr.apis.ssc.event.SscEvent;
import com.aispeech.asr.apis.ssc.logger.BigDataLogger;
import com.aispeech.asr.apis.ssc.metrics.Prometheus;
import com.aispeech.asr.comm.utils.ShellUtils;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class ProgressManager {
    private final AppConfig config;
    private final MongoOperations operations;
    private final ApplicationContext applicationContext;
    private final Set<TaskDoc> tasksInProgress = new CopyOnWriteArraySet<>();

    public ProgressManager(AppConfig config, MongoOperations operations, ApplicationContext applicationContext) {
        this.config = config;
        this.operations = operations;
        this.applicationContext = applicationContext;
    }

    public Set<TaskDoc> getTasksInProgress() {
        return tasksInProgress;
    }

    public List<String> taskIdListInProgress() {
        List<String> taskIds = new ArrayList<>();
        List<TaskDoc> tasksExpired = new ArrayList<>();

        tasksInProgress.forEach(taskDoc -> {
            if (taskIsTimeout(taskDoc)) {
                tasksExpired.add(taskDoc);
            } else {
                taskIds.add(taskDoc.getId());
            }
        });

        tasksExpired.forEach(taskDoc -> {
            finish(taskDoc, new HashMap<String, Object>() {{
                put("errid", 1);
                put("error", "timeout");
            }});
        });

        return taskIds;
    }

    public List<TaskProgressInfo> taskInfoListInProgress() {
        List<TaskProgressInfo> taskInfos = new ArrayList<>();
        tasksInProgress.forEach(taskDoc -> taskInfos.add(new TaskProgressInfo(taskDoc.getId(), taskDoc.getProgress())));
        return taskInfos;
    }

    public Optional<TaskDoc> taskDocInProgress(String taskId) {
        return tasksInProgress.stream().filter(taskDoc -> taskDoc.getId() == taskId).findFirst();
    }

    private boolean taskIsTimeout(TaskDoc taskDoc) {
        boolean timeout = false;

        long elpsedSeconds = (System.currentTimeMillis() - taskDoc.getUpdate_t()) / 1000;

        long duration = (long) taskDoc.getRequest().getOrDefault("duration", 0L);
        if (duration > 0) {
            timeout = (elpsedSeconds > ShellUtils.timeoutByDuration(duration));

            if (timeout)
                log.error("[ taskIsTimeout ] task: {}, duration: {}, elpsedSeconds: {}, timeout: {}", taskDoc.getId(), duration, elpsedSeconds, timeout);
        }

        return timeout;
    }

    public void syncToMongo(TaskDoc taskDoc, int progress) {
        String taskId = taskDoc.getId();

        if (progress == Progress.DOWNLOADING) {
//            log.info("env: {}, node: {}, product: {}", Prometheus.Env(), Prometheus.NodeName(), taskDoc.getProductId());
            Prometheus.GAUGE_N_CONCURRENCY_TASKs.labels(
                    Prometheus.Env(),
                    Prometheus.NodeName(),
                    taskDoc.getProductId()
            ).inc();
        }

        if (progress < Progress.FINISHED && !tasksInProgress.contains(taskId)) {
            tasksInProgress.add(taskDoc);
        } else if (progress >= Progress.FINISHED) {
            tasksInProgress.remove(taskDoc);
        }

        taskDoc.setProgress(progress);

        // 不能再往上放了。会导致连续pick 2次任务，重复执行。
        if ((System.currentTimeMillis() - taskDoc.getUpdate_t()) > 1000L) {
            taskDoc.setUpdate_t(System.currentTimeMillis());
            operations.save(taskDoc);
        }
    }

    public void finish(TaskDoc taskDoc) {
        finish(taskDoc, new HashMap<String, Object>() {{
                    put("errid", 0);
                    put("error", "");
                }}
        );
    }

    public void failed(SscEvent event, int errno, String error) {
        if (event.retryAgain(config.getMaxRetries())) {
            log.warn("[ RETRY ] task: {}, progress: {}, retry times: {}, errno: {}, error: {}", event.getTaskDoc().getId(),
                    event.getTaskDoc().getProgress(), event.getRetries(), errno, error);

//            log.info("[ metrics ]: {}", event.getTaskDoc().getMetrics());

            applicationContext.publishEvent(event);
        } else {
            finish(event.getTaskDoc(), new HashMap<String, Object>() {{
                        put("errid", errno);
                        put("error", error);
                    }}
            );

//            log.warn("[ failed 2 ] task: {}, progress: {}, retry times: {}, errno: {}, error: {}", event.getTaskDoc().getId(),
//                    event.getTaskDoc().getProgress(), event.getRetries(), errno, error);

            BigDataLogger.logTask(event.getTaskDoc());
        }
    }

    public void finish(TaskDoc taskDoc, Map<String, Object> error) {
        taskDoc.setProgress(Progress.FINISHED);
        taskDoc.setUpdate_t(System.currentTimeMillis());
        taskDoc.addMetric("completed_t");
        taskDoc.setFinished_t(Instant.now());

        Prometheus.GAUGE_N_CONCURRENCY_TASKs.labels(
                Prometheus.Env(),
                Prometheus.NodeName(),
                taskDoc.getProductId()
        ).dec();

        String errno = "0";
        if (error != null) {
            taskDoc.setError(error);
            errno = String.valueOf(error.get("errid"));
        }

        Prometheus.COUNTER_TASK_REQUESTS.labels(
                Prometheus.Env(),
                Prometheus.NodeName(),
                taskDoc.getProductId(),
                errno
        ).inc();

//        log.info("[ finish 2 ] task: {}", taskDoc);
        try {
            Object oduration = taskDoc.getRequest().getOrDefault("duration", 0);
            long duration = (oduration instanceof Integer) ? ((Integer) oduration).longValue() : (long)oduration;
//            log.info("[ finish 3 ] task: {}, duration: {}", taskDoc, duration);
            if (duration > 0) {
                long download_t = taskDoc.getMetrics().get("downloading_t");
                long finished_t = taskDoc.getUpdate_t();

                if (download_t > 0 && finished_t > download_t) {
                    Prometheus.SUMMARY_TASK_REAIL_RATE.labels(
                            Prometheus.Env(),
                            Prometheus.NodeName()
                    ).observe((finished_t - download_t) * 1.0f / Math.max(duration, 600000));

                    if (duration < 600000) {
                        Prometheus.SUMMARY_TASK_REAIL_RATE_LESS
                                .labels(Prometheus.Env(), Prometheus.NodeName())
                                .observe((finished_t - download_t) * 1.0f / duration);
                    }
                }
            }
        } catch (Exception ex) {
            log.error("[ finish ] err.", ex);
        }

//        log.info("[ finish ] id: {}, task: {}", taskDoc.getId(), taskDoc);

        operations.findAndReplace(
                query(where("_id").is(taskDoc.getId())),
                taskDoc
        );

        tasksInProgress.remove(taskDoc);
    }

    public class TaskProgressInfo {
        private String taskId;
        private int progress;

        public TaskProgressInfo(String taskId, int progress) {
            this.taskId = taskId;
            this.progress = progress;
        }

        @Override
        public String toString() {
            return String.format("Task<%s, %d>", taskId, progress);
        }
    }
}
